#include "Avalon/IO/PipedMessageReader.hpp"
#include "Avalon/IO/IOException.hpp"
#include "Avalon/IO/PipedMessageWriter.hpp"

using namespace Avalon;
using namespace Avalon::IO;
using namespace Avalon::Threading;
using namespace boost;
using namespace std;

PipedMessageReader::PipedMessageReader()
    : m_mutex(new recursive_mutex()),
      m_sizeRemaining(0),
      m_nextReadSize(0),
      m_destination(NULL),
      m_destinationBuffer(NULL),
      m_writer(NULL) {}

PipedMessageReader::~PipedMessageReader() {
  Break();
}

void PipedMessageReader::Break() {
  lock_guard<recursive_mutex> lock(*m_mutex);
  if(m_writer == NULL) {
    return;
  }
  m_writer->InternalBreak();
  InternalBreak();
}

bool PipedMessageReader::IsAtEndOfDataSource() {
  return false;
}

Async<int>::Ptr PipedMessageReader::Read(void* destination, int size) {
  lock_guard<recursive_mutex> lock(*m_mutex);
  m_destination = destination;
  m_nextReadSize = size;
  if(m_sizeRemaining != 0) {
    return NewAsync(CopyOver());
  }
  if(m_writer == NULL) {
    AsyncOwner<int>::Ptr result = AsyncOwner<int>::Create();
    result->SetException(IOException("Pipe broken."));
    return result;
  }
  m_nextReadResult = AsyncOwner<int>::Create();
  return m_nextReadResult;
}

Async<int>::Ptr PipedMessageReader::Read(Buffer* destination, int size) {
  lock_guard<recursive_mutex> lock(*m_mutex);
  m_destinationBuffer = destination;
  m_nextReadSize = size;
  if(m_sizeRemaining != 0) {
    return NewAsync(CopyOver());
  }
  if(m_writer == NULL) {
    AsyncOwner<int>::Ptr result = AsyncOwner<int>::Create();
    result->SetException(IOException("Pipe broken."));
    return result;
  }
  m_nextReadResult = AsyncOwner<int>::Create();
  return m_nextReadResult;
}

void PipedMessageReader::Connect(PipedMessageWriter* writer) {
  m_writer = writer;
}

void PipedMessageReader::InternalBreak() {
  m_writer = NULL;
  if(m_nextReadResult != NULL) {
    AsyncOwner<int>::Ptr nextReadResult = m_nextReadResult;
    m_nextReadResult.reset();
    nextReadResult->SetException(IOException("Pipe broken."));
  }
}

void PipedMessageReader::Write(const Buffer& data) {
  m_messages.push_back(data);
  if(m_messages.size() == 1) {
    m_sizeRemaining = data.GetSize();
    if(m_destination != NULL || m_destinationBuffer != NULL) {
      int sizeRead = CopyOver();
      AsyncOwner<int>::Ptr nextReadResult = m_nextReadResult;
      m_nextReadResult.reset();
      m_nextReadSize = 0;
      nextReadResult->SetResult(sizeRead);
    }
  }
}

int PipedMessageReader::CopyOver() {
  int sizeRead = min(m_nextReadSize, m_sizeRemaining);
  if(m_destinationBuffer != NULL) {
    m_destinationBuffer->Reserve(sizeRead);
    m_destination = m_destinationBuffer->GetMutableData();
  }
  const Buffer& buffer = m_messages.front();
  const char* readPosition = buffer.GetData() +
    (buffer.GetSize() - m_sizeRemaining);
  memcpy(m_destination, readPosition, sizeRead);
  m_sizeRemaining -= sizeRead;
  m_destination = NULL;
  m_destinationBuffer = NULL;
  if(m_sizeRemaining == 0) {
    m_messages.pop_front();
    if(!m_messages.empty()) {
      m_sizeRemaining = m_messages.front().GetSize();
    }
  }
  return sizeRead;
}
